package com.sfx.多线程代码题.BlockingQueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created with IntelliJ IDEA.
 * Description: 单锁实现阻塞队列
 * User: sfx
 * Date: 2023-05-10
 * Time: 11:00
 */
public class BlockingQueueSingleLock<E> implements BlockingQueue<E>{
    //头部
    private int head;
    //尾部
    private int tail;
    //队列中实际元素的个数
    private AtomicInteger size = new AtomicInteger();
    //数组
    private E[] queue;

    private int capacity;

    public BlockingQueueSingleLock(int capacity) {
        this.capacity = capacity;
        queue = (E[]) new Object[capacity];
    }

    //锁对象
    private ReentrantLock tailLock = new ReentrantLock();
    private ReentrantLock headLock = new ReentrantLock();

    //当队列满的时候进入的条件变量(tailWaits)
    private Condition tailWaits = tailLock.newCondition();
    //当队列空的时候进入的条件变量(headWaits)
    private Condition headWaits = headLock.newCondition();

    @Override
    public void offer(E e) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c ;//加入元素前的元素个数
        try{
            while(isFull()) {//offer1 offer2 offer3
                tailWaits.await();
            }
            this.queue[tail] = e;
            if(++tail == queue.length) {
                tail = 0;
            }
            c = size.getAndIncrement();
            //如果加入一个元素,仍然还有空位的时候,那么就需要当前offer线程去级联的唤醒其他的offer线程
            if(c + 1 < queue.length) {
                tailWaits.signal();
            }
        }finally {
            tailLock.unlock();
        }
        //只让第一个offer线程,去唤醒poll线程,剩下的都poll线程都让它自己去级联唤醒
        if(c == 0) {
            headLock.lock();
            try {
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c;
        long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
        try{
            while(isFull()) {
                if(nanos <=0) {
                    return false;
                }
                nanos = tailWaits.awaitNanos(nanos);//返回剩余的等待时间
            }
            queue[tail] = e;
            if(++tail == queue.length) {
                tail = 0;
            }
            c = size.getAndIncrement();
            //如果加入一个元素,仍然还有空位的时候,那么就需要当前offer线程
            //去级联的唤醒其他的offer线程
            if(c + 1 < queue.length) {
                tailWaits.signal();
            }
        }finally {
            tailLock.unlock();
        }
        //只让第一个offer线程,去唤醒poll线程,剩下的都poll线程都让它自己去级联唤醒
        if(c == 0) {
            headLock.lock();
            try {
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }
        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        headLock.lockInterruptibly();
        int c;//取走前的元素个数
        E e;
        try{
            while(isEmpty()) {//poll1 poll2 poll3
                headWaits.await();
            }
            e = queue[head];
            queue[head] = null;
            if(++head == queue.length) {
                head = 0;
            }
            c = size.getAndDecrement();
            //如果剩下的poll线程还右,那么就由poll线程自己级联唤醒
            //后序的poll线程
            if(c>1) {
                headWaits.signal();
            }
        }finally {
            headLock.unlock();
        }
        //同样的,当元素个数从满-->到达不满,也就是需要唤醒第一个offer线程
        //剩下的都由offer线程自己级联去唤醒
        if(c == queue.length) {
            tailLock.lock();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }

    private boolean isEmpty() {
        return this.size.get() == 0;
    }

    private boolean isFull() {
        return this.size.get() == this.capacity;
    }

    @Override
    public String toString() {
        return "MyBlockingQueue{" +
                "queue=" + Arrays.toString(queue);
    }
}